package serializable

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.{After, Before, Test}

class Driver {

	val conf: SparkConf = new SparkConf().setAppName("Serializ").setMaster("local[3]")
	var sc: SparkContext = null
	var outpath: String = "out"
	import util.MyPredef._

	@Before
	def init() {
		sc = new SparkContext(conf)
		outpath.delete
	}

	@After
	def after() {
		sc.stop()
	}

	@Test
	def serializ1(): Unit = {

		val word: RDD[String] = sc.textFile("in").flatMap(_.split(" "))
		val search = new Search1("h")
		val res: RDD[String] = word.filter(search.isMatch) // 对象在Driver端,search 没有序列化,报错 object not serializable
		res.foreach(println)
	}

	@Test
	def serializ2(): Unit = {

		val word: RDD[String] = sc.textFile("in").flatMap(_.split(" "))
		val search = new Search2("h")
		val res: RDD[String] = word.filter(search.isMatch) // 增加序列化，没问题
		res.foreach(println)
	}

	@Test
	def serializ3(): Unit = {

		val word: RDD[String] = sc.textFile("in").flatMap(_.split(" "))
		val search = new Search3("h")
		val res: RDD[String] = search.getMatche(word)	// 对象在Driver端,search 没有序列化,报错 object not serializable
		res.foreach(println)
	}

	@Test
	def serializ4(): Unit = {

		val word: RDD[String] = sc.textFile("in").flatMap(_.split(" "))
		val search = new Search4("h")
		val res: RDD[String] = search.getMatche(word)
		res.foreach(println)
	}


}
class Search1(query: String){

	//过滤出包含字符串的数据
	def isMatch(s: String): Boolean = {
		s.contains(query)
	}
}

class Search2(query: String) extends java.io.Serializable{

	//过滤出包含字符串的数据
	def isMatch(s: String): Boolean = {
		s.contains(query)
	}
}

class Search3(query: String) {

	//过滤出包含字符串的RDD
	def getMatche(rdd: RDD[String]): RDD[String] = {
		rdd.filter(x => x.contains(query))
	}

}

class Search4(query: String) {

	//过滤出包含字符串的RDD
	def getMatche(rdd: RDD[String]): RDD[String] = {
		val s = query
		rdd.filter(x => x.contains(s))
	}

}